package com.github.slprj.slake.net;

import com.github.slprj.slake.common.SlTime;
import com.github.slprj.slake.log.SlLog;
import com.github.slprj.slake.log.SlLogs;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.concurrent.TimeUnit;

/**
 * rpc 服务器
 *
 * @author YangXiaoLong
 * 2022/8/23 10:47
 */
public class NioServer extends NioNet implements Runnable {
    public static final SlLog log = SlLogs.getLogger(NioServer.class);
    private final NioServerConfig config;
    private ServerSocketChannel ssc;
    private Selector selector;
    private final Thread acceptThread;
    private volatile boolean stop;
    private final NioWorker[] workers;

    private static final PingProcessor PING_PROCESSOR = new PingProcessor();

    public NioServer(NioServerConfig config) {
        super(config);
        this.config = config;
        if (config.getPort() <= 0) {
            throw new IllegalArgumentException("no port");
        }
        //处理与client的连接请求
        acceptThread = new Thread(this);
        acceptThread.setName(config.getName() + "IoAccept");
        //接收消息
        workers = new NioWorker[config.getIoThreads()];
        for (int i = 0; i < workers.length; i++) {
            workers[i] = new NioWorker(nioStatus, config.getName() + "IoWork" + i, config);
        }
        register(Commands.CMD_PING, PING_PROCESSOR);
    }

    @Override
    protected void doStart() throws Exception {
        //初始化ServerSocketChannel
        initSsc();
        //初始化业务线程池
        super.initBizExecutor();
        //启动接受请求线程
        acceptThread.start();
        for (NioWorker worker : workers) {
            worker.start();
        }
    }

    public void initSsc() throws Exception {
        ssc = ServerSocketChannel.open();
        //非阻塞模式,可以做其他的事情
        ssc.configureBlocking(false);
        //1.让端口释放后立即就可以被再次使用
        ssc.setOption(StandardSocketOptions.SO_REUSEADDR, true);
        ssc.bind(new InetSocketAddress(config.getPort()));
        selector = Selector.open();
        //绑定selector
        ssc.register(selector, SelectionKey.OP_ACCEPT);
        log.info("{} listen at port {}", config.getName(), config.getPort());
    }

    @Override
    public void run() {
        while (!stop) {
            select();
        }
        //关闭
        try {
            selector.close();
            ssc.close();
            log.info("accept thread finished: {}", config.getName());
        } catch (IOException e) {
            log.error("close error. name={}, port={}", config.getName(), config.getPort(), e);
        }
    }

    private void select() {
        try {
            //阻塞等待事件 服务器启动时, 并未存在接收连接进行事件,此时NioWorker已经在另外一个线程初始完成
            this.selector.select();
            Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
            while (iterator.hasNext()) {
                SelectionKey key = iterator.next();
                iterator.remove();
                if (!key.isValid()) {
                    //记录日志
                    if (log.isDebugEnabled()) {
                        log.debug("Accept SelectionKey is invalid, name={}, port= {}"
                                , config.getName(), config.getPort());
                    }
                    continue;
                }
                //接收到消息
                if (key.isAcceptable()) {
                    SocketChannel sc = ssc.accept();
                    log.debug("accept new socket: {}", sc);
                    workers[sc.hashCode() % workers.length].newChannelAccept(sc);
                }
            }
            //关闭选择器异常
        } catch (ClosedSelectorException e) {
            log.warn("selector closed. name={}, port={}", config.getName(), config.getPort());
        } catch (Throwable e) {
            log.error("accept thread failed. name={}, port={}", config.getName(), config.getPort(), e);
        }
    }

    @Override
    protected void doStop() throws Exception {
        SlTime timeout = new SlTime(config.getCloseTimeoutMills(), TimeUnit.MILLISECONDS);
        stop = true;
        if (selector != null) {
            selector.wakeup();
        }

        for (NioWorker worker : workers) {
            worker.preStop();
        }

        // TODO: 2022/9/17 缺代码

        for (NioWorker worker : workers) {
            worker.stop();
        }
        //todo 存在的意义
        for (NioWorker worker : workers) {
            long rest = timeout.rest(TimeUnit.MILLISECONDS);
            if (rest > 0) {
                try {
                    worker.getThread().join();
                } catch (InterruptedException e) {
                    ThreadUtils.restoreInterruptStatus();
                }
            }
        }
        shutdownBizExecutor(timeout);

    }

    public static final class PingProcessor extends ReqProcessor {
        @Override
        public WriteFrame process(ReadFrame frame, SlChannel channel) {
            ByteBuffer body = (ByteBuffer) frame.getBody();
            log.info("收到消息长度 = [{}]", body.remaining());
            ByteBufferWriteFrame resp = new ByteBufferWriteFrame();
            resp.setBody((ByteBuffer) frame.getBody());
            // TODO: 2022/9/26 如果不设置,卡主
            resp.setRespCode(CmdCodes.SUCCESS);
            return resp;
        }

        @Override
        public Decoder getDecoder() {
            return ByteBufferDecoder.INSTANCE;
        }
    }
}
